package com.gitee.wsl.flow.combine

//import com.gitee.wsl.coroutines.atomic.AtomicRef
import com.gitee.wsl.ext.any.NULL_VALUE
import com.gitee.wsl.flow.F
import kotlinx.atomicfu.atomic
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.channels.onFailure
import kotlinx.coroutines.channels.onSuccess
import kotlinx.coroutines.channels.produce
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.emitAll
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.selects.select
import kotlinx.coroutines.yield
import kotlinx.coroutines.CoroutineStart
import kotlinx.coroutines.flow.FlowCollector
import kotlinx.coroutines.flow.SharingStarted
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.collect
import kotlinx.coroutines.flow.combine
import kotlinx.coroutines.flow.filter
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.scan
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.flow.take
import kotlinx.coroutines.launch



/**
 * Returns a [Flow] that skips items emitted by the source [Flow] until a second [Flow] emits a value or completes.
 *
 * @param notifier The second [Flow] that has to emit a value before the source [Flow]'s values
 * begin to be mirrored by the resulting [Flow].
 */
fun <T> Flow<T>.skipUntil(notifier: Flow<Any?>): Flow<T> = flow {
    coroutineScope {
        val shouldEmit = atomic(false)

        val job = launch(start = CoroutineStart.UNDISPATCHED) {
            notifier.take(1).collect()
            shouldEmit.value=true
        }

        collect {
            if (shouldEmit.value) {
                emit(it)
            }
        }

        job.cancel()
    }
}

/**
 * This exception is thrown when operator need no more elements from the flow.
 * This exception should never escape outside of operator's implementation.
 * This exception can be safely ignored by non-terminal flow operator if and only if it was caught by its owner
 * (see usages of [checkOwnership]).
 */
internal class ClosedException(val owner: FlowCollector<*>) :
    Exception("Flow was aborted, no more elements needed")

internal fun ClosedException.checkOwnership(owner: FlowCollector<*>) {
    if (this.owner !== owner) throw this
}


/**
 * Emits the values emitted by the source [Flow] until a [notifier] [Flow] emits a value or completes.
 *
 * @param notifier The [Flow] whose first emitted value or complete event
 * will cause the output [Flow] of [takeUntil] to stop emitting values from the source [Flow].
 */
fun <T> Flow<T>.takeUntil(notifier: Flow<Any?>): Flow<T> = flow {
    try {
        coroutineScope {
            val job = launch(start = CoroutineStart.UNDISPATCHED) {
                notifier.take(1).collect()
                throw ClosedException(this@flow)
            }

            collect { emit(it) }
            job.cancel()
        }
    } catch (e: ClosedException) {
        e.checkOwnership(this@flow)
    }
}

/**
 * This function is an alias to [skipUntil] operator.
 *
 * @see skipUntil
 */
@Suppress("NOTHING_TO_INLINE")
 inline fun <T> Flow<T>.dropUntil(notifier: Flow<Any?>): Flow<T> = skipUntil(notifier)

@Suppress("NOTHING_TO_INLINE")
inline fun <A, B> Flow<A>.withLatestFrom(other: Flow<B>): Flow<Pair<A, B>> =
    withLatestFrom(other) { a, b -> a to b }



/**
 * Maps [StateFlow] into [StateFlow] in synchronous manner without having to provide a default value manually.
 *
 * Issues in Kotlin bug tracker:
 * [#2008](https://github.com/Kotlin/kotlinx.coroutines/issues/2008)
 * [#2514](https://github.com/Kotlin/kotlinx.coroutines/issues/2514)
 * [#2631](https://github.com/Kotlin/kotlinx.coroutines/issues/2631)
 */
fun <T, R> StateFlow<T>.mapState(
    scope: CoroutineScope,
    sharingStarted: SharingStarted = SharingStarted.Eagerly,
    mapper: (T) -> R,
) =
    map { mapper(it) }
        .stateIn(scope, sharingStarted, mapper(value))

/**
 * Combines [StateFlow] into [StateFlow] in synchronous manner without having to provide a default value manually.
 *
 * Issues in Kotlin bug tracker:
 * [#2008](https://github.com/Kotlin/kotlinx.coroutines/issues/2008)
 * [#2514](https://github.com/Kotlin/kotlinx.coroutines/issues/2514)
 * [#2631](https://github.com/Kotlin/kotlinx.coroutines/issues/2631)
 */
inline fun <reified T, R> F.combineState(
    flows: Iterable<StateFlow<T>>,
    scope: CoroutineScope,
    sharingStarted: SharingStarted = SharingStarted.Eagerly,
    crossinline transform: (Array<T>) -> R,
): StateFlow<R> =
    combine(flows) { transform(it) }
        .stateIn(
            scope = scope,
            started = sharingStarted,
            initialValue = transform(flows.map { it.value }.toTypedArray())
        )

inline fun <reified T1, reified T2, R> StateFlow<T1>.combineState(
    flow: StateFlow<T2>,
    scope: CoroutineScope,
    sharingStarted: SharingStarted = SharingStarted.Eagerly,
    crossinline transform: (T1, T2) -> R,
): StateFlow<R> =
    combine(this, flow) { one, two -> transform(one, two) }
        .stateIn(
            scope = scope,
            started = sharingStarted,
            initialValue = transform(this.value, flow.value)
        )

fun <T> Flow<T>.withPrevious(): Flow<CompareValues<T>> =
    scan(CompareValues<T>()) { previous, current -> previous.combine(current) }
        .filter { it.isInitialized }

class CompareValues<T>(
    val previous: T? = null,
    private val currentNullable: T? = null,
) {
    val current: T
        get() = currentNullable ?: error("Should not be invoked")

    val isInitialized: Boolean
        get() = currentNullable != null

    fun combine(new: T): CompareValues<T> =
        CompareValues(currentNullable, new)

    operator fun component1(): T? = previous

    operator fun component2(): T = current

}


/** Repeats the Flow forever */
fun <A> Flow<A>.repeat(): Flow<A> =
    flow {
        while (true) {
            collect {
                emit(it)
            }
        }
    }